package com.atguigu.gmall.cdc;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * @Description: 使用FlinkCDC读取MySQL数据库表变化 自定义反序列化器. 实现手动指定数据格式.
 * @Author: tiancy
 * @Create: 2021/11/26
 */
public class FlinkCDC03_Custom {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall0609_realtime")
                .tableList("gmall0609_realtime.table_process")
                .username("root")
                .password("123456")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MyCustomSchema())
                .build();


        env.addSource(sourceFunction).print();

        env.execute("Print MySQL Snapshot + Binlog");
    }
}

/**
 * 如果自己不会写,完全可以参考 new StringDebeziumDeserializationSchema()的源码.
 * 自定义反序列化器,手动指定使用 FlinkCDC 监测到的数据输出格式 .
 * 1、implements DebeziumDeserializationSchema<String>
 * 2、重写 deserialize方法,并从写 toString.
 * 3、注意使用到Struct这个,需要导这个包.  import org.apache.kafka.connect.data.Struct;
 * 最终数据格式 :
 *  {"database":"gmall0609_realtime","data":{"name":"Hello","id":4,"age":30},"type":"insert","table":"t_user"}
 *  {"database":"gmall0609_realtime","data":{"name":"xxxx","id":4,"age":10},"type":"update","table":"t_user"}
 */
class MyCustomSchema implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> out) throws Exception {
        /*
            来对 sourceRecord 这个对象中的属性中,提取自己想要的东西,简化监测到的结果如下
            ConnectRecord{
              value=Struct{
                after=Struct{id=2,name=tiancy,age=18},
                source=Struct{db=gmall0609_realtime,table=t_user},
                op=c}
            }
         */
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");

        //获取数据库名
        String database = sourceStruct.getString("db");
        //获取表名
        String table = sourceStruct.getString("table");
        //操作类型的处理,这里仿照FlinkCDC中反序列化时的感觉 , insert、update、delete. 不要c、i、u、d这种.
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }

        //影响的数据行的处理
        Struct afterStruct = valueStruct.getStruct("after");
        //创建一个json对象，用于接收影响数据行
        JSONObject dataJsonObj = new JSONObject();
        if (afterStruct != null) {
            //获取afterstruct结构体中有哪些属性
            List<Field> fieldList = afterStruct.schema().fields(); // 获取表结构和字段名列表.
            //对结构体中所有属性进行遍历
            for (Field field : fieldList) {
                //获取属性名称
                String fieldName = field.name();
                //获取属性的值
                Object fieldValue = afterStruct.get(field);
                //将遍历出来的属性名和值  放到json对象中
                dataJsonObj.put(fieldName, fieldValue);
            }
        }

        //整体返回的json对象的封装
        JSONObject resJsonObj = new JSONObject();
        resJsonObj.put("database", database);
        resJsonObj.put("table", table);
        resJsonObj.put("type", type);
        resJsonObj.put("data", dataJsonObj);

        out.collect(resJsonObj.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}